package com.bw.gmall.realtime.dim.functions;

import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.bean.TableProcessDim;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.HbaseUtil;
import com.bw.gmall.realtime.common.util.JdbcUtil;
import com.bw.gmall.realtime.common.util.RedisUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

import java.util.HashMap;
import java.util.List;

public class DimSinkFunction extends RichSinkFunction<Tuple2<JSONObject, TableProcessDim>> {
    private Connection hbaseConnect;
    private Jedis jedis;

    @Override
    public void open(Configuration parameters) throws Exception {
        hbaseConnect = HbaseUtil.getHbaseConnect();
        jedis = RedisUtil.getJedis();
    }

    @Override
    public void invoke(Tuple2<JSONObject, TableProcessDim> value, Context context) throws Exception {
        // 主流维度数据，并已经过滤好字段了
        JSONObject f0 = value.f0;
        // 主流对应维度配置数据
        TableProcessDim f1 = value.f1;
        // 拿出写的数据
        JSONObject data = f0.getJSONObject("data");
        String type = f0.getString("type");

        String sinkTable = f1.getSinkTable();
        String sinkRowKey = f1.getSinkRowKey();
        String sinkRowKeyValue = data.getString(sinkRowKey);
        String sinkFamily = f1.getSinkFamily();

        if ("delete".equals(type)){
            HbaseUtil.deleteCells(hbaseConnect, Constant.HBASE_NAMESPACE,sinkTable,sinkRowKeyValue);
        }else{
            HbaseUtil.putCells(hbaseConnect,Constant.HBASE_NAMESPACE,sinkTable,sinkRowKeyValue,sinkFamily,data);
        }
        // 要保证缓存数据和Hbase一致性
        if("delete".equals(type) || "update".equals(type)){
            jedis.del(RedisUtil.getRedisKey(sinkTable,sinkRowKeyValue));
        }

    }

    @Override
    public void close() throws Exception {
        HbaseUtil.closeHBaseConn(hbaseConnect);
        RedisUtil.closeJedis(jedis);
    }
}
